package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import springfox.documentation.spring.web.json.Json;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    private MqttClient mqttClient;
    private MqttContext mqttContext;

    public MqttConsumer(MqttClient mqttClient, MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    /**
     * 连接成功 客户端mqttClient成功的建立了和emqx连接
     *
     * @param reconnect mqttClient 回调消费传递的参数 告知 方法 本次连接建立是否支持重连
     * @param serverURI 服务端emqx地址
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //客户端对象mqttClient成功连接的时候 调用的 通过客户端mqttClient实现监听订阅
        try {
            mqttClient.subscribe("/charging/check/result");
            mqttClient.subscribe("/charging/progress");
        } catch (MqttException e) {
            log.error("订阅主题失败", e);
        }
    }

    /**
     * 连接断开 mqttClient断开和服务连接 回调的方法
     * 消费者使用的一些资源 消耗程序内容 在连接断开时候 释放避免因为这个消费者无法使用消费逻辑导致程序阻塞
     *
     * @param cause the reason behind the loss of connection.
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error("连接断开:{}", cause.getMessage());
    }

    /**
     * 消费者通过mqttClient监听订阅的主题 产生了消息,消息到达了当前消费对象
     *
     * @param topic   消息主题来源 主题名称
     * @param message 消息对象
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //当emqx主题产生消息的时候 mqttClient 回调这个方法 告诉消费者消息对象和主题来源
        log.debug("主题:{},消息:{}", topic, message);
        if (topic != null && topic.equals("/charging/check/result")) {
            handleCheckResult(topic, message);
        } else if (topic != null && topic.equals("/charging/progress")) {
            //当前消费者 绑定的连接对象监听多个主题,如果消息来源是 充电进度主题,处理进度同步
            handleChargingProgress(topic, message);
        }
    }

    private void handleChargingProgress(String topic, MqttMessage message) {
        log.debug("当前消费者,主题:{},获取到消息:{},开始处理进度同步", topic, message);
        String jsonMsg = new String(message.getPayload(), StandardCharsets.UTF_8);
        ChargingProgressMsg chargingProgressMsg = JSON.parseObject(jsonMsg, ChargingProgressMsg.class);
        //判断温度 是否安全,不安全停止设备通知用户
        checkTemperature(chargingProgressMsg);
        //根据计价中心返回的数据vo对象完成后续流程
      ProgressCostVO  progressCostVO = calculate(chargingProgressMsg);
      //存储es
        savaProgress(chargingProgressMsg,progressCostVO.getTotalCost());
      //判断是否充满
        Boolean isFull = chargingProgressMsg.getIsFull();
        if (isFull){

        }else {

        }
    }

    private ProgressCostVO calculate(ChargingProgressMsg chargingProgressMsg) {
        return null;
    }

    private void savaProgress(ChargingProgressMsg chargingProgressMsg, Double totalCost) {
    }

    private void checkTemperature(ChargingProgressMsg chargingProgressMsg) {
    }

    private void handleCheckResult(String topic, MqttMessage message) {

        log.debug("当前消费者,主题:{},获取到消息:{},开始处理设备自检反馈", topic, message);
        log.debug("先抢锁");
        String msJson = new String(message.getPayload(), StandardCharsets.UTF_8);
        ChargingCheckResultMessage msg = JSON.parseObject(msJson, ChargingCheckResultMessage.class);
        String lockKey = topic + msg.getOrderNo();
        Boolean lock = mqttContext.getRedisTemplate().opsForValue().setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
        if (!lock) {
            log.debug("抢锁失败");
            return;
        }
        Boolean result = msg.getResult();
        if (result) {
            //SUCCESS
            ChargingBillSuccessPO po = new ChargingBillSuccessPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setBillStatus(1);
            po.setCreateTime(new Date());
            mqttContext.getChargingBillSuccessMapper().insert(po);
            //给设备发信息,可以充电
        } else {
            //FAIL
            ChargingBillFailPO po = new ChargingBillFailPO();
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setCreateTime(new Date());
            mqttContext.getChargingBillFailMapper().insert(po);
            //给用户发优惠劵充电失败给客户端发消息
            WebSocketMessage<String> webSocketMessage = new WebSocketMessage<>();
            webSocketMessage.setState(1);
            webSocketMessage.setMessage("fail");
            webSocketMessage.setData("B 您的订单创建失败,设备有故障,对不起,送您一张优惠券");
            mqttContext.getWebSocketServer().pushMessage(msg.getUserId(), JSON.toJSONString(webSocketMessage));
        }
    }
    /**
     * 消息投递 消息消费彻底完成
     *
     * @param token message身份标识
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息投递完成");
    }
}
